package com.wangyh8216.demo.flink.caculate;

// Flink相关导入
import org.apache.flink.api.common.eventtime.WatermarkStrategy;  // 水印策略，用于处理事件时间
import org.apache.flink.api.common.functions.MapFunction;        // Map函数接口
import org.apache.flink.api.common.functions.ReduceFunction;      // Reduce函数接口
import org.apache.flink.api.java.tuple.Tuple2;                   // 二元组数据类型
import org.apache.flink.streaming.api.datastream.DataStream;      // 数据流抽象
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; // 单输出流操作符
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; // 流执行环境
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows; // 翻滚事件时间窗口
import org.apache.flink.streaming.api.windowing.time.Time;       // 时间相关配置
import org.apache.flink.streaming.api.windowing.windows.TimeWindow; // 时间窗口
import org.apache.flink.util.Collector;                          // 数据收集器

// Java标准库导入
import java.time.Duration;                                       // 时间间隔
import java.util.ArrayList;                                      // 动态数组
import java.util.Date;                                           // 日期时间
import java.util.List;                                           // 列表接口
import java.util.Random;                                         // 随机数生成
import java.util.regex.Matcher;                                  // 正则表达式匹配器
import java.util.regex.Pattern;                                  // 正则表达式模式

public class LocalTopIpRatioJob {

    /**
     * 日志模式正则表达式
     * 用于解析Apache/Nginx风格的访问日志格式
     * 分组说明：
     * 1. (\\S+) - IP地址
     * 2. (.*?) - 时间戳
     * 3. (.*?) - HTTP请求
     * 4. (\\d+) - HTTP状态码
     * 5. (\\d+) - 响应大小
     * 6. (.*?) - Referer
     * 7. (.*?) - User-Agent
     * 注意：这里我们只关心IP字段，其他字段可以随意填写
     */
    private static final Pattern LOG_PATTERN = Pattern.compile(
            "^(\\S+) - - \\[(.*?)\\] \"(.*?)\" (\\d+) (\\d+) \"(.*?)\" \"(.*?)\".*");

    public static void main(String[] args) throws Exception {

        // 1. 构造本地测试数据 - 模拟60条访问日志
        List<String> mockLogs = new ArrayList<>();
        // 日志模板：标准Apache/Nginx访问日志格式
        String template = "192.168.%d.%d - - [04/Sep/2025:12:00:%02d +0000] \"GET / HTTP/1.1\" 200 1234 \"-\" \"curl\"";
        
        // 生成60条测试数据，时间戳从00秒到59秒
        // IP分布：192.168.0.x 与 192.168.1.x 两个网段
        for (int i = 0; i < 60; i++) {
            int a = new Random().nextInt(2);            // 生成0或1，用于选择网段(0.x或1.x)
            int b = new Random().nextInt(3);    // 生成0-2，用于IP地址的最后一位
            mockLogs.add(String.format(template, a, b, i));  // 格式化日志并添加到列表
        }

        // 2. 创建流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度为1，方便观察输出结果（生产环境应根据实际情况调整）
        env.setParallelism(1);

        // 3. 从内存集合创建数据源（Source）
        // 将mockLogs列表转换为DataStream<String>数据流
        DataStream<String> raw = env.fromCollection(mockLogs);

        // 4. 数据转换：解析日志行 -> (ip, 1L) 二元组
        // 使用map操作将每行日志转换为(IP地址, 计数1)的元组
        SingleOutputStreamOperator<Tuple2<String, Long>> parsed = raw
                .map(new MapFunction<String, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> map(String line) throws Exception {
                        // 使用正则表达式匹配日志行
                        Matcher m = LOG_PATTERN.matcher(line);
                        if (m.find()) {
                            // 提取IP地址（第一个分组）
                            String ip = m.group(1);
                            // 返回(IP, 1)元组，表示该IP访问一次
                            return Tuple2.of(ip, 1L);
                        }
                        // 如果日志格式不匹配，返回null
                        return null;
                    }
                })
                // 显式指定返回类型，避免Java类型擦除问题
                .returns(org.apache.flink.api.common.typeinfo.Types.TUPLE(org.apache.flink.api.common.typeinfo.Types.STRING, org.apache.flink.api.common.typeinfo.Types.LONG));

        // 5. 分配时间戳和水印（允许2秒乱序）
        // 为数据流分配事件时间戳和水印，用于基于事件时间的窗口计算
        SingleOutputStreamOperator<Tuple2<String, Long>> withWatermark =
                parsed.assignTimestampsAndWatermarks(
                        // 使用有界乱序水印策略，允许最多2秒的乱序
                        WatermarkStrategy.<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                                .withTimestampAssigner((event, ts) -> {
                                    // 时间戳分配器：为每条数据分配时间戳
                                    // 这里使用当前系统时间加上随机毫秒数，模拟真实的时间分布
                                    // 注意：生产环境中应该从数据中提取真实的时间戳
                                    return System.currentTimeMillis() + (long)(Math.random() * 1000);
                                }));

        // 6. 窗口聚合：10秒翻滚窗口统计每个IP的访问次数
        // 使用keyBy按IP分组，然后在10秒的翻滚窗口内进行reduce聚合
        DataStream<Tuple2<String, Long>> ipCount = withWatermark
                // 按IP地址分组（value.f0是IP字段）
                .keyBy(value -> value.f0)
                // 定义10秒的翻滚事件时间窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                // reduce聚合函数：将相同IP的计数累加
                .reduce(new ReduceFunction<Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> reduce(
                            Tuple2<String, Long> v1, Tuple2<String, Long> v2) {
                        // 保持IP不变，将计数相加
                        return Tuple2.of(v1.f0, v1.f1 + v2.f1);
                    }
                });

        // 7. 全局窗口处理：计算每个窗口内的Top-1 IP及其占比
        // 使用windowAll对所有数据进行全局窗口处理
        SingleOutputStreamOperator<String> result = ipCount
                // 定义10秒的全局翻滚窗口
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(10)))
                // 使用ProcessAllWindowFunction处理整个窗口的数据
                .process(new org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction<
                        Tuple2<String, Long>,
                        String,
                        TimeWindow>() {
                    @Override
                    public void process(Context ctx,
                                        Iterable<Tuple2<String, Long>> elements,
                                        Collector<String> out) {
                        long total = 0;          // 总访问次数
                        Tuple2<String, Long> top = null;  // Top-1 IP及其计数
                        
                        // 遍历窗口内的所有元素
                        for (Tuple2<String, Long> e : elements) {
                            total += e.f1;       // 累加总访问次数
                            // 找出访问次数最多的IP
                            if (top == null || e.f1 > top.f1) {
                                top = e;
                            }
                        }
                        
                        // 如果有数据，计算Top IP的占比并输出结果
                        if (top != null) {
                            // 计算Top IP的访问占比
                            double ratio = total == 0 ? 0.0 : top.f1 * 1.0 / total;
                            // 格式化输出结果
                            out.collect(String.format(
                                    "window=%s~%s, top-ip=%s, count=%d, total=%d, ratio=%.2f%%",
                                    new Date(ctx.window().getStart()),  // 窗口开始时间
                                    new Date(ctx.window().getEnd()),    // 窗口结束时间
                                    top.f0,                             // Top IP地址
                                    top.f1,                             // Top IP访问次数
                                    total,                              // 总访问次数
                                    ratio * 100));                      // 占比百分比
                        }
                    }
                });

        // 8. 输出结果：将处理结果打印到控制台
        result.print();

        // 9. 执行流处理任务
        // 启动Flink作业执行，作业名称为"Local Top IP Ratio"
        env.execute("Local Top IP Ratio");
    }
}